package com.atguigu.gmall.realtime.app.dws;

import com.atguigu.gmall.realtime.app.func.KeywordUDTF;
import com.atguigu.gmall.realtime.bean.KeywordBean;
import com.atguigu.gmall.realtime.common.GmallConstant;
import com.atguigu.gmall.realtime.util.MyClickHouseUtil;
import com.atguigu.gmall.realtime.util.MyKafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author Felix
 * @date 2022/8/5
 * 今天知识点总结
 *      Clickhouse副本
 *      Clickhouse分片集群
 *
 * 来源关键词聚合统计----FlinkSQL
 *      环境准备
 *      从kakfa中读取数据创建动态表  并在建表的时候指定Watermark以及提取事件时间字段
 *      对表中的数据进行过滤
 *      使用自定函数进行分词 并和表中的其它字段进行连接
 *      分组、开窗、聚合计算
 *      将动态表转换为流
 *      将流中的数据写到clickhouse表中(基本的实现)
 */
public class DwsTrafficSourceKeywordPageViewWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //1.3 指定表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //1.4 注册函数类
        tableEnv.createTemporarySystemFunction("ik_analyze", KeywordUDTF.class);

        //TODO 2.检查点相关设置(略)

        //TODO 3.从页面日志主题中 dwd_traffic_page_log 读取数据创建动态表  指定Watermark以及提取事件时间字段
        String topic = "dwd_traffic_page_log";
        String groupId = "dws_traffic_keyword_group";
        tableEnv.executeSql("create table page_log(\n" +
            " common map<string,string>,\n" +
            " page map<string,string>,\n" +
            " ts bigint,\n" +
            " rowtime as TO_TIMESTAMP(FROM_UNIXTIME(ts/1000)),\n" +
            " WATERMARK FOR rowtime AS rowtime - INTERVAL '3' SECOND\n" +
            ") " + MyKafkaUtil.getKafkaDDL(topic,groupId));

        // tableEnv.executeSql("select * from page_log").print();

        //TODO 4.从页面日志中过滤出搜索行为
        Table searchTable = tableEnv.sqlQuery("select\n" +
            " page['item'] fullword,\n" +
            " rowtime\n" +
            "from page_log where page['last_page_id'] = 'search' \n" +
            "and page['item_type']='keyword' and page['item'] is not null");

        tableEnv.createTemporaryView("search_table",searchTable);
        // tableEnv.executeSql("select * from search_table").print();

        //TODO 5.使用分词函数进行分词     并将分词结果和表中其它字段进行连接
        Table joinedTable = tableEnv.sqlQuery("SELECT keyword,rowtime FROM search_table,LATERAL TABLE(ik_analyze(fullword)) t(keyword)");
        tableEnv.createTemporaryView("joined_table",joinedTable);
        // tableEnv.executeSql("select * from joined_table").print();

        //TODO 6.分组、开窗、聚合计算
        Table resTable = tableEnv.sqlQuery("select \n" +
            " DATE_FORMAT(TUMBLE_START(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') stt,\n" +
            " DATE_FORMAT(TUMBLE_END(rowtime, INTERVAL '10' SECOND), 'yyyy-MM-dd HH:mm:ss') edt,\n" +
            " '" + GmallConstant.KEYWORD_SEARCH + "' source,\n" +
            " keyword ,\n" +
            " count(*) keyword_count,\n" +
            " UNIX_TIMESTAMP()*1000 ts\n" +
            "from joined_table group by TUMBLE(rowtime, INTERVAL '10' SECOND),keyword");

        //TODO 7.将动态表转化为流
        DataStream<KeywordBean> keywordBeanDS = tableEnv.toAppendStream(resTable, KeywordBean.class);

        keywordBeanDS.print(">>>>>");

        //TODO 8.将流中的数据写到Clickhouse对应的表中
        keywordBeanDS.addSink(
            MyClickHouseUtil.getSinkFunction("insert into dws_traffic_source_keyword_page_view_window values(?,?,?,?,?,?)")
        );

        env.execute();
    }
}
